package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.TradeCourseOrderBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;
import java.math.BigDecimal;

public class DwsTradeCourseOrderWindowSyncCache extends BaseApp {
    public static void main(String[] args) {
        new DwsTradeCourseOrderWindowSyncCache().start(
                11111,
                4,
                "dws_trade_course_order_window",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //kafkaStrDS.print();
        // TODO: 2024/10/21 1.读数据  +  类型转换  +去空处理
        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String string, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        //  JSONObject jsonObject = JSON.parseObject(string);
                        if (StringUtils.isNotEmpty(string)) {
                            JSONObject jsonObject = JSON.parseObject(string);
                            collector.collect(jsonObject);
                        }
                    }
                }
        );
        // jsonObj.print();
        // TODO: 2024/10/21 2.去重!!! 状态存储前一条数据   因为在做dwd订单明细表时生成3条数据!!!!
        // 2.1  先按订单明细id,唯一主键,进行分组
        KeyedStream<JSONObject, String> keyByJsonObj = jsonObj.keyBy(jsonObject -> jsonObject.getString("id"));
        // keyByJsonObj.print();
        //keyByJsonObj.process(
        //        new ProcessFunction<JSONObject, JSONObject>() {
        //            // TODO: 2024/10/21 开状态!!!
        //            private ValueState valueState;
        //            // 状态初始化!!!
        //            @Override
        //            public void open(Configuration parameters) throws Exception {
        //                // 声明状态描述器
        //                ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<>("valueState", JSONObject.class);
        //                // 赋值
        //                valueState = getRuntimeContext().getState(valueStateDescriptor);
        //            }
        //            // 流中数据处理
        //            @Override
        //            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
        //                // 把度量值取出来,然后取反!!!
        //                String originAmount = keyByJsonObj.get("origin_amount");
        //            }
        //        }
        //);
        SingleOutputStreamOperator<JSONObject> distinctDS = keyByJsonObj.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    // TODO: 2024/10/21 开状态!!!
                    ValueState<JSONObject> lastData;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor = new ValueStateDescriptor<>("lastData", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(5)).build());
                        lastData = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        // TODO: 2024/10/21 先从状态里获取上条到来的数据值
                        //Object value = lastData.value();
                        JSONObject v1 = lastData.value();
                        if (v1 != null) {
                            // TODO: 2024/10/21 说明是重复数据!!!  准备取反
                            // TODO: 2024/10/21 取出度量值 !!!
                            String originAmount = v1.getString("origin_amount");
                            String couponReduce = v1.getString("coupon_reduce");
                            String finalAmount = v1.getString("final_amount");

                            v1.put("origin_amount", "-" + originAmount);
                            v1.put("coupon_reduce", "-" + couponReduce);
                            v1.put("final_amount", "-" + finalAmount);

                            // TODO: 2024/10/21 把状态值对象往外发!!!
                            collector.collect(v1);
                        }
                        // TODO: 2024/10/21 更新状态    然后把新状态往外发!!!
                        lastData.update(jsonObject);
                        collector.collect(jsonObject);
                    }
                }
        );
        // distinctDS.print();
        // TODO: 2024/10/21 3.第二次类型转换   jsonObj -> 封装成统计用的实体bean类
        SingleOutputStreamOperator<TradeCourseOrderBean> beanDS = distinctDS.map(
                new MapFunction<JSONObject, TradeCourseOrderBean>() {
                    @Override
                    public TradeCourseOrderBean map(JSONObject jt) throws Exception {
                        String orderDetailId = jt.getString("id");
                        String courseId = jt.getString("course_id");
                        String userId = jt.getString("user_id");
                        String courseName = jt.getString("course_name");
                        BigDecimal originAmount = jt.getBigDecimal("origin_amount");
                        BigDecimal couponReduce = jt.getBigDecimal("coupon_reduce");
                        BigDecimal finalAmount = jt.getBigDecimal("final_amount");
                        long ts = jt.getLong("ts") * 1_000;

                        // TODO: 2024/10/21 建造者模式启动
                        TradeCourseOrderBean bean1 = TradeCourseOrderBean.builder()
                                .orderDetailId(orderDetailId)
                                .userId(userId)
                                .courseId(courseId)
                                .courseName(courseName)
                                .originalAmount(originAmount)
                                .couponReduceAmount(couponReduce)
                                .finalAmount(finalAmount)
                                .ts(ts)
                                .build();
                        return bean1;
                    }
                }
        );
        //beanDS.print();
        // TODO: 2024/10/21 4.水位线
        SingleOutputStreamOperator<TradeCourseOrderBean> watermarks = beanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeCourseOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeCourseOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeCourseOrderBean bean, long l) {
                                        return bean.getTs();
                                    }
                                }
                        )
        );

        // TODO: 2024/10/21 5.按course粒度进行分组
        KeyedStream<TradeCourseOrderBean, String> courseIdKeyedDS = watermarks.keyBy(TradeCourseOrderBean::getCourseId);
        // courseIdKeyedDS.print();

        // TODO: 2024/10/21 6.开窗
        WindowedStream<TradeCourseOrderBean, String, TimeWindow> windowDS = courseIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));

        // TODO: 2024/10/21 7.聚合处理
        SingleOutputStreamOperator<TradeCourseOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeCourseOrderBean>() {
                    // TODO: 2024/10/22 聚合度量字段!!!!
                    @Override
                    public TradeCourseOrderBean reduce(TradeCourseOrderBean v1, TradeCourseOrderBean v2) throws Exception {
                        v1.setOriginalAmount(v1.getOriginalAmount().add(v2.getOriginalAmount()));
                        v1.setCouponReduceAmount(v1.getCouponReduceAmount().add(v2.getCouponReduceAmount()));
                        v1.setFinalAmount(v1.getFinalAmount().add(v2.getFinalAmount()));
                        return v1;
                    }
                },
                new WindowFunction<TradeCourseOrderBean, TradeCourseOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeCourseOrderBean> iterable, Collector<TradeCourseOrderBean> collector) throws Exception {
                        TradeCourseOrderBean orderBean = iterable.iterator().next();
                        orderBean.setStt(DateFormatUtil.tsToDateTime(window.getStart()));
                        orderBean.setEdt(DateFormatUtil.tsToDateTime(window.getEnd()));
                        orderBean.setCurDate(DateFormatUtil.tsToDate(window.getStart()));
                        // TODO: 2024/10/22  改完对象往外发!!!!
                        collector.collect(orderBean);
                    }
                }
        );
        // reduceDS.print();


        // TODO: 2024/10/21 8.维度关联      使用优化方法  旁路缓存 + 异步io +模板设计方法     额外得做一个异步连接函数function
        // TODO: 2024/10/22 1. 多层连续关联  课程id > 课程信息表（course_info） >学科id > 科目表（base_subject_info） > 分类id >分类表（base_category_info）
        SingleOutputStreamOperator<TradeCourseOrderBean> bean1 = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeCourseOrderBean>() {
                    @Override
                    public void addDims(TradeCourseOrderBean bean, JSONObject dimJsonObj) {
                        bean.setSubjectId(dimJsonObj.getString("subject_id"));
                        // bean.setCategoryName(dimJsonObj.getString("category_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getRowKey(TradeCourseOrderBean bean) {
                        return bean.getCourseId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        // bean1.print();

        SingleOutputStreamOperator<TradeCourseOrderBean> bean2 = AsyncDataStream.unorderedWait(
                bean1,
                new DimAsyncFunction<TradeCourseOrderBean>() {
                    @Override
                    public void addDims(TradeCourseOrderBean bean, JSONObject dimJsonObj) {
                        bean.setCategoryId(dimJsonObj.getString("category_id"));
                        bean.setSubjectName(dimJsonObj.getString("subject_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_subject_info";
                    }

                    @Override
                    public String getRowKey(TradeCourseOrderBean bean) {
                        return bean.getSubjectId();
                    }
                },
                90,
                TimeUnit.SECONDS
        );
       //  bean2.print();

        SingleOutputStreamOperator<TradeCourseOrderBean> bean3 = AsyncDataStream.unorderedWait(
                bean2,
                new DimAsyncFunction<TradeCourseOrderBean>() {
                    @Override
                    public void addDims(TradeCourseOrderBean bean, JSONObject dimJsonObj) {
                        bean.setCategoryName(dimJsonObj.getString("category_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category_info";
                    }

                    @Override
                    public String getRowKey(TradeCourseOrderBean bean) {
                        return bean.getCategoryId();
                    }
                },
                120,
                TimeUnit.SECONDS
        );
        // bean3.print();

        // TODO: 2024/10/22 2. 用户id > 用户表（user_info）
        SingleOutputStreamOperator<TradeCourseOrderBean> bean4 = AsyncDataStream.unorderedWait(
                bean3,
                new DimAsyncFunction<TradeCourseOrderBean>() {
                    @Override
                    public void addDims(TradeCourseOrderBean bean, JSONObject dimJsonObj) {
                        bean.setUserName(dimJsonObj.getString("real_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_user_info";
                    }

                    @Override
                    public String getRowKey(TradeCourseOrderBean bean) {
                        return bean.getUserId();
                    }
                },
                60,
                TimeUnit.SECONDS
        );
         bean4.print();


        // TODO: 2024/10/22 在doris中建表   然后把关联结果写入!!!
        // bean4.print();
        bean4.map(new BeanToJsonStrMapFunction<>())
               .sinkTo(FlinkSinkUtil.getDorisSink("dws_trade_course_order_window"));

    }
}
